package comsume.listener;

import comsume.config.MySink;
import comsume.model.User;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.stereotype.Component;

/**
 * @author xiaohong
 * @version 1.0
 * @date 2021/12/15 0015 12:28
 * @description 接收普通消息
 */
@Component
@Slf4j
public class MessageListener {

    /**
     * 测试集群模式
     * @param message
     */
    @StreamListener(MySink.CLUSTER_INPUT)
    public void onClusterMessage(@Payload User message) {
        log.info("[集群模式][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
    }

    /**
     * 测试广播模式
     * @param message
     */
    @StreamListener(MySink.RADIO_INPUT)
    public void onRadioMessage(@Payload User message) {
        log.info("[广播模式][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
    }

    /**
     * 测试消息重试
     * @param message
     */
    @StreamListener(MySink.RETRY_INPUT)
    public void onTryMessage(@Payload User message) {
        log.info("[消息重试][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
        // <X> 注意，此处抛出一个 RuntimeException 异常，模拟消费失败
        throw new RuntimeException("我就是故意抛出一个异常");
    }

    /**
     * 测试顺序消息
     * @param message
     */
    @StreamListener(MySink.ORDER_INPUT)
    public void onOrderMessage(@Payload User message) {
        log.info("[顺序消息][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
    }

    /**
     * 测试根据tag过滤消息 Broker级别
     * @param message
     */
    @StreamListener(MySink.FILTER_INPUT)
    public void onBrokerFilterMessage(@Payload User message) {
        log.info("[根据tag过滤消息 Broker级别][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
    }

    /**
     * 测试根据tag过滤消息 Consumer级别
     * 只需要使用 @StreamListener 注解的 condition 属性，设置消息满足指定 Spring EL 表达式的情况下，才进行消费
     * 注意，被过滤掉的消息，后续是无法被消费掉了，效果和消费成功是一样的
     * @param message
     */
    @StreamListener(value = MySink.FILTER_INPUT, condition = "headers['rocketmq_TAGS'] == 'yunai'")
    public void onConsumerFilterMessage(@Payload User message) {
        log.info("[根据tag过滤消息 Consumer级别][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
    }

    /**
     * 测试事务消息
     * @param message
     */
    @StreamListener(MySink.TRANSACTIONAL_INPUT)
    public void onTransactionalMessage(@Payload User message) {
        log.info("[事务消息][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
    }

    /**
     * 局部的异常处理：通过订阅指定错误 Channel
     * RETRY-TOPIC-01.try-consumer-group-RETRY-TOPIC-01
     * 对应者该binding配置里的destination.group
     * @param errorMessage
     */
    @ServiceActivator(inputChannel = "RETRY-TOPIC-01.retry-consumer-group-RETRY-TOPIC-01.errors")
    public void handleError(ErrorMessage errorMessage) {
        log.error("[handleError][payload：{}]", ExceptionUtils.getRootCauseMessage(errorMessage.getPayload()));
        log.error("[handleError][originalMessage：{}]", errorMessage.getOriginalMessage());
        log.error("[handleError][headers：{}]", errorMessage.getHeaders());
    }

    /**
     * 全局的异常处理：通过订阅全局错误 Channel
     * 在全局和局部异常处理都定义的情况下，错误消息仅会被符合条件的局部错误异常处理。如果没有符合条件的，错误消息才会被全局异常处理。
     * 不过要注意，如果异常处理方法成功，没有重新抛出异常，会认定为该消息被消费成功，所以就不会进行消费重试
     * @param errorMessage
     */
    @StreamListener(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME) // errorChannel
    public void globalHandleError(ErrorMessage errorMessage) {
        log.error("[globalHandleError][payload：{}]", ExceptionUtils.getRootCauseMessage(errorMessage.getPayload()));
        log.error("[globalHandleError][originalMessage：{}]", errorMessage.getOriginalMessage());
        log.error("[globalHandleError][headers：{}]", errorMessage.getHeaders());
    }

}
